package cn.getech.data.development.sink.jdbc.retract

import java.lang
import java.sql.{Connection, DriverManager, PreparedStatement}

import cn.getech.data.development.sink.jdbc.config.CustomTableConfig
import org.apache.flink.api.java.tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.jdbc.utils.JdbcUtils
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.types.Row


class FlinkStreamJDBCSink(conf: CustomTableConfig) extends RichSinkFunction[tuple.Tuple2[lang.Boolean, Row]] {

  private var conn: Connection = null
  private var stmt: PreparedStatement = null

  override def open(parameters: Configuration): Unit = {
    val config = conf.getConfig()
    Class.forName(config.driver._2)
    conn = DriverManager.getConnection(config.url._2, config.username._2, config.password._2)
    stmt = conn.prepareStatement(conf.sqlParser)
  }

  override def invoke(value: tuple.Tuple2[lang.Boolean, Row]): Unit = {
    if (value.f0) {
      val ints = conf.param_class_type2Ints
      for (i <- 0 until conf.getParamSize) {
        JdbcUtils.setField(stmt, ints(i), value.f1.getField(i), i)
      }
      stmt.execute()
    }
  }

  override def close(): Unit = {
    if (stmt != null || !stmt.isClosed) {
      stmt.close
    }
    if (conn != null || !conn.isClosed) {
      conn.close
    }
  }

}
